package com.lzwk.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lzwk.app.function.StatLogParse;
import com.lzwk.config.JobConfig;
import com.lzwk.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author: CC
 * @Date: 2022/3/7 19:53
 * @descprtion 听课日志分流
 */
public class StatLogShunt {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        JobConfig prod = new JobConfig("prod", env);
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(prod.statLogTopic, prod.statLogGroupId,prod.statLogKafkaServers));

        //脏数据流
        OutputTag<String> outputTag = new OutputTag<String>("dirty") {
        };
        //微课
        OutputTag<String> wkTag = new OutputTag<String>("wk") {
        };
        //哪吒
        OutputTag<String> nzTag = new OutputTag<String>("nz") {
        };



        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) {
                try {
                    //解析日志
                    JSONObject jsonObject = JSON.parseObject(value);
                    JSONObject resultJsonObject = StatLogParse.run(jsonObject);
                    out.collect(resultJsonObject);
                } catch (Exception e) {
                    ctx.output(outputTag, value);
                }
            }
        });

        SingleOutputStreamOperator<JSONObject> jsonObjFilterDS = jsonObjDS.filter(
                new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        String uri = value.getString("uri");
                        return "/__t.gif".equals(uri);
                    }
                }
        );


        SingleOutputStreamOperator<String> pageDS = jsonObjFilterDS.process(new ProcessFunction<JSONObject, String>() {
                                                                          @Override
                                                                          public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                                                                              // 根据不同project写入dwd
                                                                              Object project = value.getOrDefault("project","");
                                                                              if (project.equals("nezha")) {
                                                                                  ctx.output(nzTag, value.toJSONString());
                                                                              } else {
                                                                                  ctx.output(wkTag, value.toJSONString());
                                                                              }
                                                                          }
                                                                      }
        );
        //打印脏数据
        DataStream<String> dirtyDS = jsonObjDS.getSideOutput(outputTag);
        DataStream<String> nzDS = pageDS.getSideOutput(nzTag);
        DataStream<String> wkDS = pageDS.getSideOutput(wkTag);

//        dirtyDS.print("dirtyDS>>>>>>>>>>>");
//        nzDS.print("nzDS>>>>>>>>>>>");
//        wkDS.print("wkDS>>>>>>>>>>>");
        dirtyDS.addSink(MyKafkaUtil.getKafkaProducer("dirty_stat_log",prod.statLogKafkaServers));
        nzDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_nezha_stat_log",prod.statLogKafkaServers));
        wkDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_weike_stat_log",prod.statLogKafkaServers));
        env.execute("stat_log_shunt");
    }
}
